package javapns.utils;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

public class ApnsUtils {

    static final Logger logger = Logger.getLogger(ApnsUtils.class);

    private static final byte[] FAM_MAIN = Bytes.toBytes("main");
    private static final byte[] COL_IOS_TOKEN = Bytes.toBytes("ios");
    private static final byte[] COL_ENABLE = Bytes.toBytes("enable");
    
    private static final int MAX_BLOCK_SIZE = 1000;

    private static String FILE_PATH = "/data/notification/invalid-tokens.txt";

    private static ReadWriteLock rwlock = new ReentrantReadWriteLock();

    public static void writeInvalidTokens(List<String> tokens) {
        rwlock.writeLock().lock();
        File path = new File(FILE_PATH);
        FileOutputStream fo = null;
        OutputStreamWriter osw = null;

        try {
            fo = new FileOutputStream(path, true);
            osw = new OutputStreamWriter(fo, "utf-8");
            for (String token : tokens) {
                osw.write(token + "\n");
            }
            osw.flush();
        } catch (Exception e) {

        } finally {
            if (fo != null)
                try {
                    fo.close();
                } catch (Exception ex) {
                }
            rwlock.writeLock().unlock();
        }
    }

    private static List<String> readInvalidTokens() {
        rwlock.readLock().lock();
        BufferedReader br = null;
        List<String> tokens = Lists.newArrayList();
        Map<String, Boolean> map = Maps.newHashMap();
        try {
            String token;
            br = new BufferedReader(new FileReader(FILE_PATH));
            while ((token = br.readLine()) != null) {
                if (!Strings.isNullOrEmpty(token)) {
                    if (!map.containsKey(token))
                        tokens.add(token);
                    map.put(token, true);
                }
            }
        } catch (Exception e) {

        } finally {
            rwlock.readLock().unlock();
            if (br != null)
                try {
                    br.close();
                } catch (Exception ex) {

                }
        }
        return tokens;
    }

    private static Map<String, String> getUIDfromToken(String tableName, List<String> invalidTokens) {
        Map<String, String> res = Maps.newHashMap();
        try {
            Map<String, Boolean> map = Maps.newHashMap();
            for (String token : invalidTokens) {
                map.put(token, true);
            }
            HTable table = new HTable(tableName);
            Scan scan = new Scan();
            scan.addColumn(FAM_MAIN, COL_IOS_TOKEN);
            scan.addColumn(FAM_MAIN, COL_ENABLE);
            ResultScanner rs = table.getScanner(scan);
            try {
                for (Result r : rs) {
                    boolean enable = Bytes.toBoolean(r.getValue(FAM_MAIN, COL_ENABLE));
                    if (enable) {
                        String userId = Bytes.toString(r.getRow());
                        String token = Bytes.toString(r.getValue(FAM_MAIN, COL_IOS_TOKEN));
                        if (!Strings.isNullOrEmpty(token)) {
                            if (map.containsKey(token))
                                res.put(userId, token);
                        }
                    }
                }
            } finally {
                table.close();
            }
        } catch (Exception e) {

        }
        return res;
    }

    public static void removeInvalidToken(String tableName) {
        logger.debug("Remove invalid tokens from hbase");
        try {
            List<String> invalidTokens = readInvalidTokens();
            if (invalidTokens.size() == 0)
                return;
            Map<String, String> rows = getUIDfromToken(tableName, invalidTokens);
            List<Delete> list = Lists.newArrayList();
            logger.debug("Found " + invalidTokens.size() + " tokens that must be remove");
            System.out.println("removed " + invalidTokens.size() + " invalid tokens");
            HTable table = new HTable(tableName);
            try {
                int n = rows.size();
                for (String rowKey : rows.keySet()) {
                    logger.debug("  remove userId " + rowKey + " with token "
                            + rows.get(rowKey) + "");
                    byte[] row = Bytes.toBytes(rowKey);
                    Delete delete = new Delete(row);
                    n--;
                    list.add(delete);
                    if (list.size() == MAX_BLOCK_SIZE || n == 0) {
                        if (list.size() > 0)
                            table.delete(list);
                        list = new ArrayList<Delete>();
                    }
                }
            } finally {
                table.close();
                File tokensFile = new File(FILE_PATH);
                if (tokensFile.exists())
                    tokensFile.delete();
            }
        } catch (Exception e) {
        }
    }

}
